/*
   Copyright (C) 2006
   Andrew Madigan

This file is part of nhttpd.

nhttpd is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2, or (at your option)
any later version.

nhttpd is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
General Public License for more details.

You should have received a copy of the GNU General Public License
along with nhttpd; see the file COPYING.  If not, write to the
Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
02110-1301 USA.

Linking this library statically or dynamically with other modules is
making a combined work based on this library.  Thus, the terms and
conditions of the GNU General Public License cover the whole
combination.

As a special exception, the copyright holders of this library give you
permission to link this library with independent modules to produce an
executable, regardless of the license terms of these independent
modules, and to copy and distribute the resulting executable under
terms of your choice, provided that you also meet, for each linked
independent module, the terms and conditions of the license of that
module.  An independent module is a module which is not derived from
or based on this library.  If you modify this library, you may extend
this exception to your version of the library, but you are not
obligated to do so.  If you do not wish to do so, delete this
exception statement from your version.
*/

#include <nhttpd/queue.h>
#include <nhttpd/log.h>
#include <errno.h>
#if NH_THREADS >= 0
#include <pthread.h>
#endif
#if NH_XMQ >= 0
#include <sys/msg.h>
#endif

#include <sys/time.h>
#include <time.h>

#ifndef NH_GQ_DETECT
#if NH_XMQ > 0 && NH_THREADS > 0
#define GQ_XMQ_Create GQ_Create
#define GQ_XMQ_SetCallback GQ_SetCallback
#define GQ_XMQ_Enqueue GQ_Enqueue
#define GQ_XMQ_Dequeue GQ_Dequeue
#else
#define GQ_PQ_Create GQ_Create
#define GQ_PQ_SetCallback GQ_SetCallback
#define GQ_PQ_Enqueue GQ_Enqueue
#define GQ_PQ_Dequeue GQ_Dequeue
#endif
#endif

typedef struct {
  long mtype;
  void* data;
} GQ_XMQ_Msg;

#if NH_XMQ >= 0
void GQ_XMQ_Free(void* _self) {
  GroupQueue* self = _self;
  msgctl(self->data.mqid, IPC_RMID, NULL);
  free(self);
}

Class GQ_XMQ_Class = {GQ_XMQ_Free};

GroupQueue* GQ_XMQ_Create(int maxqueued, void* ptr) {
  GroupQueue* self = malloc(sizeof(GroupQueue));
  self->super.methods = &GQ_XMQ_Class;
  self->data.mqid = msgget(IPC_PRIVATE, IPC_CREAT);
  self->maxqueued = maxqueued;

  if (maxqueued >= 0) {
	 struct msqid_ds* buf = malloc(sizeof(struct msqid_ds));
	 msgctl(self->data.mqid, IPC_STAT, buf);
	 buf->msg_qbytes = maxqueued * sizeof(void*);
	 msgctl(self->data.mqid, IPC_SET, buf);
	 free(buf);
  }

  return self;
}

void GQ_XMQ_SetCallback(GroupQueue* self, int type, NHQ_Handler callback) {
}

bool GQ_XMQ_Enqueue(GroupQueue* self, void* value, int type, NHQ_Flag flag) {
  GQ_XMQ_Msg* msg = malloc(sizeof(GQ_XMQ_Msg));
  msg->mtype = type;
  msg->data = value;
  return msgsnd(self->data.mqid, msg, sizeof(GQ_XMQ_Msg), flag == NHQ_BLOCK ? 0 
	  : IPC_NOWAIT) == 0 ? true : false;
}

void* GQ_XMQ_Dequeue(GroupQueue* self, int type, NHQ_Flag flag) {
	GQ_XMQ_Msg* msg = malloc(sizeof(GQ_XMQ_Msg));
	if (msgrcv(self->data.mqid, msg, sizeof(GQ_XMQ_Msg), type, flag == NHQ_BLOCK 
		? 0 : IPC_NOWAIT) == -1) {
		free(msg);
		return NULL;
	}
	else {
		void* rv = msg->data;
		free(msg);
		return rv;
	}
}
#endif

#if NH_XMQ <= 0 || NH_THREADS <= 0
void GQ_PQ_Free(void* _self) {
  GroupQueue* self = _self;
  O_Free(self->data.pqh.table);
  free(self);
}

Class GQ_PQ_Class = {GQ_PQ_Free};

GroupQueue* GQ_PQ_Create(int maxqueued, void* ptr) {
  GroupQueue* self = malloc(sizeof(GroupQueue));
  self->super.methods = &GQ_PQ_Class;
  self->maxqueued = maxqueued;

  self->data.pqh.table = HT_Create(true, true);
  self->data.pqh.ptr = ptr;
  return self;
}

void GQ_PQ_SetCallback(GroupQueue* self, int type, NHQ_Handler callback) {
  HT_Put(self->data.pqh.table, (char*) &type, 
	  (char*) PQ_Create(self->maxqueued, callback, self->data.pqh.ptr));
}

bool GQ_PQ_Enqueue(GroupQueue* self, void* value, int type, NHQ_Flag flag) {
  PoolQueue* queue = HT_Get(self->data.pqh.table, (char*) &type);
  return PQ_Enqueue(queue, value, flag);
}

void* GQ_PQ_Dequeue(GroupQueue* self, int type, NHQ_Flag flag) {
  PoolQueue* queue = HT_Get(self->data.pqh.table, (char*) &type);
  return PQ_Dequeue(queue, flag);
}

#endif

#ifndef NH_PQ_DETECT
#if NH_THREADS > 0
#define PQ_TQ_Create PQ_Create
#define PQ_TQ_Enqueue PQ_Enqueue
#define PQ_TQ_Dequeue PQ_Dequeue
#define PQ_TQ_TimedEnqueue PQ_TimedEnqueue
#define PQ_TQ_TimedDequeue PQ_TimedDequeue
#else
#define PQ_CB_Create PQ_Create
#define PQ_CB_Enqueue PQ_Enqueue
#define PQ_CB_Dequeue PQ_Dequeue
#define PQ_DEF_TimedEnqueue PQ_TimedEnqueue
#define PQ_DEF_TimedDequeue PQ_TimedDequeue
#endif
#endif

#if NH_TIMEOUTS <= 0 || NH_THREADS <= 0 || NH_PMQ <= 0
bool PQ_DEF_TimedEnqueue(PoolQueue* self, long nanos, void* value, 
	NHQ_Flag altflag) {
  errno = 0;
  PQ_Enqueue(self, value, altflag);
  return errno == 0;
}

void* PQ_DEF_TimedDequeue(PoolQueue* self, long nanos, NHQ_Flag altflag) {
  return PQ_Dequeue(self, altflag);
}
#endif

#if NH_THREADS >= 0
void PQ_TQ_Free(void* _self) {
	PoolQueue* self = _self;
	PQ_ThreadQueue* queue = self->data.queue;
	pthread_mutex_destroy(&queue->mutex);
	pthread_cond_destroy(&queue->producedCondition);
	pthread_cond_destroy(&queue->consumedCondition);
	O_Free(queue->list);
	free(queue);
	free(self);
}

Class PQ_TQ_Class = {PQ_TQ_Free};

PoolQueue* PQ_TQ_Create(int maxqueued, NHQ_Handler callback, void* ptr) {
  PoolQueue* self = malloc(sizeof(PoolQueue));
  self->super.methods = &PQ_TQ_Class;
  PQ_ThreadQueue* queue = self->data.queue = malloc(sizeof(PQ_ThreadQueue));
  pthread_mutex_init(&queue->mutex, NULL);
  pthread_cond_init(&queue->producedCondition, NULL);
  pthread_cond_init(&queue->consumedCondition, NULL);
  queue->maxqueued = maxqueued;
  queue->list = L_Create();
  return self;
}

bool PQ_TQ_Enqueue(PoolQueue* self, void* value, NHQ_Flag flag) {
  PQ_ThreadQueue* queue = self->data.queue;
  pthread_mutex_lock(&queue->mutex);
  if (flag == NHQ_NOBLOCK && queue->list->length >= queue->maxqueued) {
	  pthread_mutex_unlock(&queue->mutex);
	  return false;
  }
  while (queue->list->length >= queue->maxqueued) {
	 pthread_cond_wait(&queue->consumedCondition, &queue->mutex);
  }
  pthread_mutex_unlock(&queue->mutex);
  L_Add(queue->list, value);
   pthread_cond_signal(&(queue->producedCondition));
  return true;
}

void* PQ_TQ_Dequeue(PoolQueue* self, NHQ_Flag flag) {
  PQ_ThreadQueue* queue = self->data.queue;
  pthread_mutex_lock(&(queue->mutex));
  if (flag == NHQ_NOBLOCK && queue->list->length == 0) {
	  pthread_mutex_unlock(&queue->mutex);
	  return NULL;
  }
  while(queue->list->length == 0)
	 pthread_cond_wait(&(queue->producedCondition), &(queue->mutex));
 	void* rv = L_Dequeue(queue->list);
	pthread_mutex_unlock(&queue->mutex);
 	pthread_cond_signal(&(queue->consumedCondition));
  return rv;
}

struct timespec* PQ_createTimespec(long nanos) {
	struct timeval* systime = malloc(sizeof(struct timeval));
	gettimeofday(systime, NULL);
	struct timespec* rv = malloc(sizeof(struct timespec));
	rv->tv_sec = systime->tv_sec + nanos / NANOS_PER_SEC;
	rv->tv_nsec = systime->tv_usec * 1000 + nanos % NANOS_PER_SEC;
	free(systime);
	return rv;
}

bool PQ_TQ_TimedEnqueue(PoolQueue* self, long nanos, void* value, 
	NHQ_Flag altflag) {
	PQ_ThreadQueue* queue = self->data.queue;
	pthread_mutex_lock(&(queue->mutex));
  if (queue->list->length >= queue->maxqueued) {
	 struct timespec* time = PQ_createTimespec(nanos);
	 pthread_cond_timedwait(&(queue->consumedCondition), &queue->mutex, time);
	 free(time);
	 if (queue->list->length >= queue->maxqueued) {
		return false;
	 }
  }
  L_Add(queue->list, value);
  pthread_mutex_unlock(&(queue->mutex));
  pthread_cond_signal(&(queue->producedCondition));
  return true;
}

void* PQ_TQ_TimedDequeue(PoolQueue* self, long nanos, NHQ_Flag altflag) {
  PQ_ThreadQueue* queue = self->data.queue;
  pthread_mutex_lock(&(queue->mutex));
  if (queue->list->length == 0) {
	 struct timespec* time = PQ_createTimespec(nanos);
	 int er = pthread_cond_timedwait(&(queue->producedCondition), &(queue->mutex), time);
	 if (er == EINVAL) {
		 NH_log(LOG_CRIT, "%m: Invalid Timespec");
	 }
	 free(time);
	 if (queue->list->length == 0) {
		 pthread_mutex_unlock(&(queue->mutex));
		return NULL;
	 }
  }
  void* rv = L_Dequeue(queue->list);
  pthread_mutex_unlock(&(queue->mutex));
 	pthread_cond_signal(&(queue->consumedCondition));
  return rv;
}
#endif

#if NH_THREADS <= 0
PoolQueue* PQ_CB_Create(int maxqueued, NHQ_Handler callback, void* ptr) {
	PoolQueue* self = malloc(sizeof(PoolQueue));
	self->data.cb.callback = callback;
	self->data.cb.ptr = ptr;
	return self;
}

bool PQ_CB_Enqueue(PoolQueue* self, void* value, NHQ_Flag flag) {
	self->data.cb.callback(self->data.cb.ptr, value);
	return true;
}

void* PQ_CB_Dequeue(PoolQueue* self, NHQ_Flag flag) {
	return NULL;
}
#endif

void NHQ_init() {
#if NH_THREADS == 0
	if (!NHD_THREADS) {
		GQ_Create = GQ_PQ_Create;
		GQ_SetCallback = GQ_PQ_SetCallback;
		GQ_Enqueue = GQ_PQ_Enqueue;
		GQ_Dequeue = GQ_PQ_Dequeue;
		PQ_Create = PQ_CB_Create;
		PQ_Enqueue = PQ_CB_Enqueue;
		PQ_Dequeue = PQ_CB_Dequeue;
		PQ_TimedEnqueue = PQ_DEF_TimedEnqueue;
		PQ_TimedDequeue = PQ_DEF_TimedDequeue;
		return;
	}
	else {
		PQ_Create = PQ_TQ_Create;
		PQ_Enqueue = PQ_TQ_Enqueue;
		PQ_Dequeue = PQ_TQ_Dequeue;
		PQ_TimedEnqueue = PQ_TQ_TimedEnqueue;
		PQ_TimedDequeue = PQ_TQ_TimedDequeue;
	}
#endif

#if NH_XMQ == 0
	if (!NHD_XMQ) {
		GQ_Create = GQ_PQ_Create;
		GQ_SetCallback = GQ_PQ_SetCallback;
		GQ_Enqueue = GQ_PQ_Enqueue;
		GQ_Dequeue = GQ_PQ_Dequeue;
	}
	else {
		GQ_Create = GQ_XMQ_Create;
		GQ_SetCallback = GQ_XMQ_SetCallback;
		GQ_Enqueue = GQ_XMQ_Enqueue;
		GQ_Dequeue = GQ_XMQ_Dequeue;
	}
#endif

	
}
